package com.study.chapter12;

import com.study.entity.LoginEvent;
import com.study.entity.OrderEvent;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.functions.TimedOutPartialMatchHandler;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * @Description:
 * @Author: LiuQun
 * @Date: 2022/8/28 17:49
 */
public class OrderTimeoutDetectExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 1.获取订单数据流
        SingleOutputStreamOperator<OrderEvent> orderStream = env.fromElements(
                new OrderEvent("user_1", "order_1", "create", 1000L),
                new OrderEvent("user_2", "order_2", "create", 2000L),
                new OrderEvent("user_1", "order_1", "modify", 10 * 1000L),
                new OrderEvent("user_1", "order_1", "pay", 60 * 1000L),
                new OrderEvent("user_2", "order_3", "create", 10 * 60 * 1000L),
                new OrderEvent("user_2", "order_3", "pay", 20 * 60 * 1000L)
        ).assignTimestampsAndWatermarks(
                WatermarkStrategy.<OrderEvent>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner((SerializableTimestampAssigner<OrderEvent>) (element, recordTimestamp) -> element.timestamp)
        );

        //2.定义模式：创建订单后且在15min内支付的模式
        Pattern<OrderEvent, OrderEvent> pattern = Pattern.<OrderEvent>begin("create")
                .where(new SimpleCondition<OrderEvent>() {
                    @Override
                    public boolean filter(OrderEvent value) throws Exception {
                        return StringUtils.equals(value.eventType, "create");
                    }
                })
                .followedBy("pay")
                .where(new SimpleCondition<OrderEvent>() {
                    @Override
                    public boolean filter(OrderEvent value) throws Exception {
                        return StringUtils.equals(value.eventType, "pay");
                    }
                })
                .within(Time.minutes(15));

        //3.将模式应用到订单数据流上
        PatternStream<OrderEvent> patternStream = CEP.pattern(orderStream.keyBy(order -> order.orderId), pattern);

        //4.定义一个侧输出流标签
        OutputTag<String> timeoutTag = new OutputTag<String>("timeout"){};

        //5.将完全匹配和超时部分匹配的复杂事件提取出来，进行处理
        SingleOutputStreamOperator<String> result = patternStream.process(new OrderPayMatch());

        //打印输出
        result.print("payed");
        result.getSideOutput(timeoutTag).print("timeout");

        env.execute();
    }

    public static class OrderPayMatch extends PatternProcessFunction<OrderEvent,String> implements TimedOutPartialMatchHandler<OrderEvent>{

        @Override
        public void processMatch(Map<String, List<OrderEvent>> match, Context ctx, Collector<String> out) throws Exception {
            //获取当前的支付事件
            OrderEvent orderEvent = match.get("pay").get(0);
            out.collect("用户"+orderEvent.userId+"订单"+orderEvent.orderId+"已支付！");
        }

        @Override
        public void processTimedOutMatch(Map<String, List<OrderEvent>> match, Context ctx) throws Exception {
            //获取当前超时未支付的事件
            OrderEvent createEvent = match.get("create").get(0);
            OutputTag<String> timeoutTag = new OutputTag<String>("timeout"){};
            ctx.output(timeoutTag,"用户"+createEvent.userId+"订单"+createEvent.orderId+"超时未支付！");
        }
    }
}
